package com.atguigu.flink.datastreamapi.dirstribute;

import com.google.common.base.Function;
import org.apache.flink.api.common.functions.Partitioner;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created by Smexy on 2023/4/3
 */
public class Demo2_CustomPartitioner
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> ds = env.socketTextStream("hadoop102", 8888);


        KeySelector<String, Integer> k1 = new KeySelector<String, Integer>()
        {

            @Override
            public Integer getKey(String value) throws Exception {
                return Integer.valueOf(value);
            }
        };

        KeySelector<String, Integer> k3 =  (String s) ->  Integer.valueOf(Integer.parseInt(s, 10)) ;

        //以上写法和 Integer类的 valueOf()的实现是一样的，不要重复造轮子，直接拿来用
        KeySelector<String, Integer> k2 = Integer::valueOf;
            /*
                自定义分区

                partitionCustom(
              Partitioner<K> partitioner：  如何根据key，返回数据所要发送的通道号。
              KeySelector<T, K> keySelector : 从 T类型中获取 key
             )
             */
            /*
                numPartitions: 是下游的通道数
                返回的结果是下游的通道索引,从0开始。
                    方法返回值的最大值 只能是  numPartitions - 1，不能等于或超过 numPartitions
             */
        ds.partitionCustom(
             // 将整数 / 10 ，如果结果 < 当前下游的通道数，就发送到结果对应的通道。否则就发送到最后一个通道！
                // Partitioner<K>: k是数据中 key的类型
            // 返回的通道号  介于 [0,numPartitions-1]
             (Integer key, int numPartitions) -> {
                int index = key / 10;
                return index >= numPartitions ? numPartitions - 1 : index;
            },

              k3  )
          .print();

        /*
            Integer::valueOf 返回一个  KeySelector<T, K> keySelector的对象。

                方法引用，是lamda表达式的一个特殊形式

                完整写法
         */


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
